package com.ferry.transport.netty.consumer;

import com.ferry.common.util.concurrent.NamedThreadFactory;
import com.ferry.transport.api.Connection;
import com.ferry.transport.api.ConnectionManager;
import com.ferry.transport.api.Directory;
import com.ferry.transport.api.UnResolvedAddress;
import com.ferry.transport.api.channel.WrappedChannelGroup;
import com.ferry.transport.api.config.TcpConfig;
import com.ferry.transport.api.config.TcpOption;
import com.ferry.transport.api.domain.Protocol;
import com.ferry.transport.netty.channel.NettyChannelGroup;
import com.google.common.collect.Maps;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.EventLoopGroup;
import io.netty.util.HashedWheelTimer;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadFactory;

/**
 * Created by daiyong
 */
public abstract class NettyConnector implements Connector<Connection> {

	private static final Logger logger = LoggerFactory.getLogger(NettyConnector.class);

	protected Protocol protocol;

	/**
	 * 时间轮算法
	 */
	protected HashedWheelTimer timer = new HashedWheelTimer(new NamedThreadFactory("connector.timer", true));

	private final ConcurrentMap<UnResolvedAddress, WrappedChannelGroup> addressGroups = Maps.newConcurrentMap();

	private Map<String, CopyOnWriteArrayList<WrappedChannelGroup>> directoryGroupMap = Maps.newConcurrentMap();

	/**
	 * 链接管理器
	 */
	private final ConnectionManager connectionManager = new ConnectionManager();

	private Bootstrap bootstrap;
	private EventLoopGroup worker;
	private int nWorkers;

	private ConsumerProcessor processor;

	public NettyConnector(Protocol protocol) {
		this(protocol, Runtime.getRuntime().availableProcessors() << 1);
	}

	public NettyConnector(Protocol protocol, int nWorkers) {
		this.protocol = protocol;
		this.nWorkers = nWorkers;
	}

	protected void init() {
		ThreadFactory workerFactory = new DefaultThreadFactory("ferry.connnector", Thread.MAX_PRIORITY);
		worker = initEventLoopGroup(nWorkers, workerFactory);

		bootstrap = new Bootstrap().group(worker);

		TcpConfig config = config();
		config.setOption(TcpOption.IO_RATIO, 100);

		doInit();
	}

	protected abstract void doInit();

	@Override
	public Protocol protocol() {
		return protocol;
	}

	@Override
	public ConsumerProcessor processor() {
		return processor;
	}

	@Override
	public void setProcessor(ConsumerProcessor processor) {
		setProcessor(this.processor = processor);
	}

	@Override
	public WrappedChannelGroup group(UnResolvedAddress address) {
		WrappedChannelGroup group = addressGroups.get(address);
		if (group == null) {
			WrappedChannelGroup newGroup = channelGroup(address);
			group = addressGroups.putIfAbsent(address, newGroup);
			if (group == null) {
				group = newGroup;
			}
		}
		return group;
	}

	protected WrappedChannelGroup channelGroup(UnResolvedAddress address) {
		return new NettyChannelGroup(address);
	}

	@Override
	public Collection<WrappedChannelGroup> groups() {
		return addressGroups.values();
	}

	@Override
	public boolean addChannelGroup(Directory directory, WrappedChannelGroup group) {
		CopyOnWriteArrayList<WrappedChannelGroup> groupList = directory(directory);
		boolean added = false;
		if (null == groupList) {
			CopyOnWriteArrayList<WrappedChannelGroup> newGroupList = new CopyOnWriteArrayList<>();
			newGroupList.addIfAbsent(group);
			groupList = directoryGroupMap.putIfAbsent(directory.directoryString(), newGroupList);
			if (null == groupList) {
				added = true;
			}
		}

		return added;
	}

	@Override
	public boolean removeChannelGroup(Directory directory, WrappedChannelGroup group) {
		CopyOnWriteArrayList<WrappedChannelGroup> groups = directory(directory);
		boolean removed = groups.remove(group);
		return removed;
	}

	@Override
	public CopyOnWriteArrayList<WrappedChannelGroup> directory(Directory directory) {
		return directoryGroupMap.get(directory.directoryString());
	}

	@Override
	public boolean isDirectoryAvailable(Directory directory) {
		CopyOnWriteArrayList<WrappedChannelGroup> groups = directory(directory);
		if (!CollectionUtils.isEmpty(groups)) {
			for (WrappedChannelGroup g : groups) {
				if (g.isAvailable()) {
					return true;
				}
			}
		}
		return false;
	}

	@Override
	public Map<String, CopyOnWriteArrayList<WrappedChannelGroup>> directoryGroup() {
		return directoryGroupMap;
	}

	@Override
	public ConnectionManager connectionManager() {
		return connectionManager;
	}

	@Override
	public void shutdownGracefully() {
		connectionManager.cancelAllAutoReconnect();
		worker.shutdownGracefully().syncUninterruptibly();
		timer.stop();
		if (processor != null) {
			processor.shutdown();
		}
	}

	protected void setOptions() {
		TcpConfig config = config();
		setIoRatio(config.getOption(TcpOption.IO_RATIO));
	}

	public abstract void setIoRatio(int workerIoRatio);

	protected abstract EventLoopGroup initEventLoopGroup(int nThreads, ThreadFactory tFactory);

	protected Bootstrap bootstrap() {
		return bootstrap;
	}

}
